package com.javaspeak.concurrency.nio.niolet.server;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.LockSupport;

import org.apache.log4j.Logger;

/**
 * @author John Dickerson
 */
public class MessageProcessingThread extends Thread {

    public Logger logger = Logger.getLogger( MessageProcessingThread.class );

    private Charset charset = Charset.forName( "UTF-8" );
    private CharsetDecoder charsetDecoder = charset.newDecoder();
    private CharsetEncoder charsetEncoder = charset.newEncoder();
    private ConcurrentLinkedQueue<Message> messageConcurrentLinkedQueue;
    private ConcurrentLinkedQueue<Message> messageResponseConcurrentLinkedQueue;
    private MessageResponseThread messageResponseThread;
    private volatile boolean keepRunning = true;
    private ServerNiolet serverNiolet;


    private void processMessage( Message message ) {

        try {
            CharBuffer charBufferIn =
                charsetDecoder.decode( message.getMessageByteBuffer() );

            String messageStr = charBufferIn.toString();

            // logger.debug( messageStr );

            String response = serverNiolet.processRequest( messageStr );

            if ( response !=  null ){

                 ByteBuffer messageOutByteBuffer =
                     charsetEncoder.encode( CharBuffer.wrap( response ) );

                 Message messageOut =
                     new Message( messageOutByteBuffer,
                             message.getSelectionKey() );

                 messageResponseConcurrentLinkedQueue.add( messageOut );

                 LockSupport.unpark( messageResponseThread );
            }
        }
        catch( CharacterCodingException e ){

            logger.error( "Could not process message", e );
        }
    }


    /**
     * Constructor
     *
     * @param messageConcurrentLinkedQueue
     * @param messageResponseConcurrentLinkedQueue
     */
    public MessageProcessingThread(
        ConcurrentLinkedQueue<Message> messageConcurrentLinkedQueue,
            ConcurrentLinkedQueue<Message> messageResponseConcurrentLinkedQueue,
                MessageResponseThread messageResponseThread,
                    ServerNiolet serverNiolet ){

        this.messageConcurrentLinkedQueue = messageConcurrentLinkedQueue;
        this.messageResponseConcurrentLinkedQueue = messageResponseConcurrentLinkedQueue;
        this.messageResponseThread = messageResponseThread;
        this.serverNiolet = serverNiolet;
    }


    public void gracefullyStop(){

        this.keepRunning = false;
    }


    public void run() {

        Message message;

        while ( keepRunning ){

            message = messageConcurrentLinkedQueue.poll();

            if ( message == null ){

                LockSupport.park();
            }
            else {

                processMessage( message );
            }
        }
    }
}
